package com.chianghao.nio.core;

import com.chianghao.nio.core.handle.ChannelContext;
import com.chianghao.nio.core.handle.ChannelReadThread;
import com.chianghao.nio.core.handle.ChannelWriteThread;
import com.chianghao.nio.core.handle.http.HttpRequestKey;
import com.chianghao.nio.core.handle.http.HttpRoute;

import java.io.IOException;
import java.net.Socket;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author ChiangHao
 * @date 2022/10/2 10:21
 * @Description:
 */
public class NioSubReactor implements Runnable {

    public final Selector selector;

    public final ReentrantLock mutex;


    Map<HttpRequestKey, HttpRoute> routeMap;

    public Map<HttpRequestKey, HttpRoute> getRouteMap() {
        return routeMap;
    }

    public void setRouteMap(Map<HttpRequestKey, HttpRoute> routeMap) {
        this.routeMap = routeMap;
    }

    public NioSubReactor(Map<HttpRequestKey, HttpRoute> routeMap) throws IOException {
        selector = Selector.open();
        mutex = new ReentrantLock();
        this.routeMap = routeMap;
    }


    public void register(SocketChannel channel) {
        if (!mutex.hasQueuedThreads()) {
            // 没有线程等待锁表示select 阻塞中，如果阻塞中则获取获取锁进行注册操作
            mutex.lock();
            try{
                this.selector.wakeup();
                registerToSelector(channel);
            }finally {
                mutex.unlock();
            }
        }else{
            registerToSelector(channel);
        }
    }


    private static void closeChannel(SocketChannel channel) {
        if (channel == null) {
            return;
        }
        Socket socket = channel.socket();
        if (socket != null) {
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        try {
            channel.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void registerToSelector(SocketChannel channel){
        // 将其注入selector中
        SelectionKey key = null;
        try {
            channel.configureBlocking(false);
            key = channel.register(this.selector, SelectionKey.OP_READ);
        } catch (Exception e) {
            e.printStackTrace();
            closeChannel(channel);
            if (key != null) {
                key.cancel();
            }
        }
    }

    @Override
    public void run() {
        for (; ; ) {
            int num =0 ;
            boolean isError = false;
            try {
                num = selector.select();
            } catch (IOException e) {
                e.printStackTrace();
                isError = true;
            }
            mutex.lock();
            try {
                if(isError || num == 0){
                    continue;
                }
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                if (selectionKeys.size() == 0) {
                    continue;
                }
                Iterator<SelectionKey> it = selectionKeys.iterator();
                while (it.hasNext()) {
                    SelectionKey key = it.next();
                    it.remove();
                    SocketChannel channel = (SocketChannel) key.channel();
                    if (key.isValid()) {
                        ChannelContext channelContext = null;
                        if(key.attachment()!=null){
                            channelContext = (ChannelContext)key.attachment();
                        }
                        if(channelContext==null){
                            channelContext = new ChannelContext();
                            key.attach(channelContext);
                        }
                        if (key.isReadable()) {
                            if(!channelContext.getIsRead()){
                                synchronized (key){
                                    if (!channelContext.getIsRead()) {
                                        channelContext.setIsRead(true);
                                        key.attach(channelContext);
                                        NioThreadPool.pool.execute(new ChannelReadThread(this,key,channel));
                                    }
                                }
                            }
                        } else if (key.isWritable()) {
                            if (!channelContext.getIsWrite()) {
                                synchronized (key){
                                    if (!channelContext.getIsWrite()) {
                                        channelContext.setIsWrite(true);
                                        key.attach(channelContext);
                                        NioThreadPool.pool.execute(new ChannelWriteThread(key,channel));
                                    }
                                }
                            }
                        }
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                mutex.unlock();
            }
        }
    }
}
